package com.foreveross.fetl.core.domain.single.repository.impl;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Collection;
import java.util.Date;
import java.util.List;
import java.util.Map;

import org.apache.log4j.Logger;
import org.springframework.stereotype.Repository;

import com.foreveross.fetl.common.vo.FetlErrorMessage;
import com.foreveross.fetl.core.domain.single.PlaneInfoJsonEntity;
import com.foreveross.fetl.core.domain.single.SinglePlaneInfoEntity;
import com.foreveross.fetl.core.domain.single.repository.ISinglePlaneInfoRepository;
import com.foreveross.fetl.util.DataBaseFetlUtils;
import com.foreveross.fetl.util.DateUtils;
import com.foreveross.fetl.util.FetlErrorUtils;
import com.foreveross.fetl.util.JsonUtil;
import com.foreveross.fetl.util.RandomUtils;
import com.foreveross.fetl.util.StringUtils;
@Repository("singlePlaneInfoRepository")
public class SinglePlaneInfoRepositoryImpl implements
		ISinglePlaneInfoRepository {
	
	private static Logger log = Logger.getLogger("sinPlanRep");
	
	@Override
	public void create(SinglePlaneInfoEntity entity) throws SQLException {
		StringBuffer insetSql =new StringBuffer(" insert into planeInfo(");
		List<Object> conditionVals = new ArrayList<Object>();
		if(null!=entity.getSourceRowkey()){
			insetSql.append("SOURCE_ROWKEY,");
			conditionVals.add(entity.getSourceRowkey());
		}
		if(null!=entity.getAttachHbaseKey()){
			insetSql.append("ATTACH_HBASE_KEY,");
			conditionVals.add(entity.getAttachHbaseKey());
		}
		if(null!=entity.getCreateTime()){
			insetSql.append("CREATE_TIME,");
			conditionVals.add(entity.getCreateTime());
		}
		if(null!=entity.getGrabDate()){
			insetSql.append("GRAB_DATE,");
			conditionVals.add(entity.getGrabDate());
		}
		if(null!=entity.getGrabChannelId()){
			insetSql.append("GRAB_CHANNEL_ID,");
			conditionVals.add(entity.getGrabChannelId());
		}
		if(null!=entity.getGrabChannelName()){
			insetSql.append("GRAB_CHANNEL_NAME,");
			conditionVals.add(entity.getGrabChannelName());
		}
		if(null!=entity.getCarrierKey()){
			insetSql.append("CARRIER_KEY,");
			conditionVals.add(entity.getCarrierKey());
		}
		if(null!=entity.getCarrierName()){
			insetSql.append("CARRIER_NAME,");
			conditionVals.add(entity.getCarrierName());
		}
		if(null!=entity.getCarrierFullName()){
			insetSql.append("CARRIER_FULL_NAME,");
			conditionVals.add(entity.getCarrierFullName());
		}
		if(null!=entity.getFlightNo()){
			insetSql.append("FLIGHTNO,");
			conditionVals.add(entity.getFlightNo());
		}
		if(null!=entity.getActuallyFlightNo()){
			insetSql.append("ACTUALLY_FLIGHT_NO,");
			conditionVals.add(entity.getActuallyFlightNo());
		}
		if(null!=entity.getFlightType()){
			insetSql.append("FLIGHT_TYPE,");
			conditionVals.add(entity.getFlightType());
		}
		if(null!=entity.getFlightDate()){
			insetSql.append("FLIGHT_DATE,");
			conditionVals.add(entity.getFlightDate());
		}
		if(null!=entity.getStartTime()){
			insetSql.append("START_TIME,");
			conditionVals.add(entity.getStartTime());
		}
		if(null!=entity.getEndTime()){
			insetSql.append("END_TIME,");
			conditionVals.add(entity.getEndTime());
		}
		if(null!=entity.getFromCity()){
			insetSql.append("FROM_CITY,");
			conditionVals.add(entity.getFromCity());
		}
		if(null!=entity.getFromCityName()){
			insetSql.append("FROM_CITY_NAME,");
			conditionVals.add(entity.getFromCityName());
		}
		if(null!=entity.getToCity()){
			insetSql.append("TO_CITY,");
			conditionVals.add(entity.getToCity());
		}
		if(null!=entity.getToCityName()){
			insetSql.append("TO_CITY_NAME,");
			conditionVals.add(entity.getToCityName());
		}
		if(null!=entity.getAreaCode()){
			insetSql.append("AREA_CODE,");
			conditionVals.add(entity.getAreaCode());
		}
		if(null!=entity.getAreaName()){
			insetSql.append("AREA_NAME,");
			conditionVals.add(entity.getAreaName());
		}
		if(null!=entity.getTotalHighestPrice()){
			insetSql.append("TOTAL_HIGHEST_PRICE,");
			conditionVals.add(entity.getTotalHighestPrice());
		}
		if(null!=entity.getTotalLowestPrice()){
			insetSql.append("TOTAL_LOWEST_PRICE,");
			conditionVals.add(entity.getTotalLowestPrice());
		}
		if(null!=entity.getAgentName()){
			insetSql.append("AGENT_NAME,");
			conditionVals.add(entity.getAgentName());
		}
		if(null!=entity.getTotalHighestTaxesPrice()){
			insetSql.append("TOTAL_HIGHEST_TAXES_PRICE,");
			conditionVals.add(entity.getTotalHighestTaxesPrice());
		}
		if(null != entity.getTotalLowestTaxesPrice()){
			insetSql.append("TOTAL_LOWEST_TAXES_PRICE,");
			conditionVals.add(entity.getTotalLowestTaxesPrice());
		}
		if(null != entity.getSumHighestPrice()){
			insetSql.append("SUM_HIGHEST_PRICE,");
			conditionVals.add(entity.getSumHighestPrice());
		}
		if(null != entity.getSumLowestPrice()){
			insetSql.append("SUM_LOWEST_PRICE,");
			conditionVals.add(entity.getSumLowestPrice());
		}
		
		Connection conn =null;
		PreparedStatement stmt = null;  
	    ResultSet rs = null;
		try {
				conn=DataBaseFetlUtils.getConnection();
			 	conn.setAutoCommit(false); 
			 	
				 int id =0;
			 	 PlaneInfoJsonEntity planeInfoJsonEntity=new PlaneInfoJsonEntity();
				 String jsonString=null;
				 if(null!=entity.getAgents() && entity.getAgents().size()>0 ){
						jsonString =JsonUtil.getJSONString(entity.getAgents());
						planeInfoJsonEntity.setAgentsJson(jsonString);
				 }
				 if(null!=entity.getCabins()&&entity.getCabins().size()>0){
						jsonString =JsonUtil.getJSONString(entity.getCabins());
						planeInfoJsonEntity.setCabinsJson(jsonString);
				 } 
				 if(null!=entity.getTransits()&& entity.getTransits().size()>0){
					 jsonString =JsonUtil.getJSONString(entity.getTransits());
					 planeInfoJsonEntity.setTransitsJson(jsonString);
				 }
				 if(null!=entity.getStopOvers()&& entity.getStopOvers().size()>0){
					 jsonString =JsonUtil.getJSONString(entity.getStopOvers());
					 planeInfoJsonEntity.setStopOversJson(jsonString);
				 }
				 
				 String jsonSql=" insert into planeinfo_json(AGENTS_JSON,CABINS_JSON,TRANSITS_JSON,STOPOVERS_JSON) values(?,?,?,?) ";
				 stmt=conn.prepareStatement(jsonSql, Statement.RETURN_GENERATED_KEYS);
				 stmt.setObject(1, planeInfoJsonEntity.getAgentsJson());
				 stmt.setObject(2, planeInfoJsonEntity.getCabinsJson());
				 stmt.setObject(3, planeInfoJsonEntity.getTransitsJson());
				 stmt.setObject(4, planeInfoJsonEntity.getStopOversJson());
				 stmt.executeUpdate(); 
				 rs = stmt.getGeneratedKeys();  
			        if (rs.next()) {  
			            id = rs.getInt(1);  
			        } 
			     if(id>0){
			    	 insetSql.append("JSON_ID,");
					 conditionVals.add(id);	
						
				}
			     
				StringBuffer sql=new StringBuffer(insetSql.substring(0, insetSql.length()-1));
				String sql2=null;
				
				sql.append(") values( ");
				
				if(null!=conditionVals && !conditionVals.isEmpty() && conditionVals.size()>0){
					for(int i=0;i<conditionVals.size();i++){
						sql.append("?,");
					}
					 sql2=sql.substring(0,sql.length()-1)+")";
					 stmt=conn.prepareStatement(sql2);
					 for (int i = 0; i < conditionVals.size(); i++) {  
			                stmt.setObject(i + 1, conditionVals.get(i));  
			            } 
					 stmt.executeUpdate(); 
					
				}
				
				 conn.commit();
				
		} catch (SQLException e) {
				try {
					conn.rollback();
				} catch (SQLException e1) {
					e1.printStackTrace();
				}
		    throw e;
		}finally{
		
			if (rs != null) { // 关闭记录集  
	            try {  
	                rs.close();  
	            } catch (SQLException e) {  
	                e.printStackTrace();  
	            }  
	        }  
	        if (stmt != null) { // 关闭声明  
	            try {  
	                stmt.close();  
	            } catch (SQLException e) {  
	                e.printStackTrace();  
	            }  
	        }  
	    	if(null!=conn){
				try {
					conn.close();
				} catch (SQLException e) {
					e.printStackTrace();
					}
				}
		}
	}

	@Override
	public void update(SinglePlaneInfoEntity entity) {
		// TODO Auto-generated method stub

	}

	@Override
	public void delete(SinglePlaneInfoEntity entity) throws SQLException {
		Connection conn =null;
		PreparedStatement stmt = null; 
		String deletejsonSql= " delete from planeinfo_json where ID=? ";
		String deleteSql=" delete from planeInfo where ID = ? ";
		 try {
			conn = DataBaseFetlUtils.getConnection();
			conn.setAutoCommit(false);
			if (null != entity.getJsonId() && entity.getJsonId()>0) {
				stmt = conn.prepareStatement(deletejsonSql);
				stmt.setObject(1, entity.getJsonId());
				stmt.execute();
			}
			stmt = conn.prepareStatement(deleteSql);
			stmt.setObject(1, entity.getId());
			stmt.execute();
			conn.commit();	
		} catch (SQLException e) {
			try {
				conn.rollback();
			} catch (SQLException e1) {
				throw e1;
			}
			throw e;
		}finally{
	        if (stmt != null) { // 关闭声明  
	            try {  
	                stmt.close();  
	            } catch (SQLException e) {  
	                e.printStackTrace();  
	            }  
	        }  
	    	if(null!=conn){
				try {
					conn.close();
				} catch (SQLException e) {
					e.printStackTrace();
					}
				}
		}
		

	}
	
	@Override
	public void delete(String hbaseKey,String fromCity,String toCity,String flightDate,String flightNo)throws SQLException{
		Connection conn =null;
		PreparedStatement stmt = null; 
		String deletejsonSql = "DELETE FROM planeinfo_json  WHERE  planeinfo_json.ID=( SELECT  JSON_ID FROM planeinfo p WHERE "
				+ " p.ATTACH_HBASE_KEY=? AND p.FROM_CITY=? AND  p.TO_CITY=? AND FLIGHT_DATE=? AND  p.FLIGHTNO=? )";

		String deleteSql = " DELETE FROM planeinfo  WHERE ATTACH_HBASE_KEY=? AND FROM_CITY=? AND "
				+ " TO_CITY=? AND FLIGHT_DATE=? AND FLIGHTNO=? ";
		 try {
			conn = DataBaseFetlUtils.getConnection();
			conn.setAutoCommit(false);
			stmt = conn.prepareStatement(deletejsonSql);
			stmt.setObject(1, hbaseKey);
			stmt.setObject(2, fromCity);
			stmt.setObject(3, toCity);
			stmt.setObject(4, flightDate);
			stmt.setObject(5, flightNo);
			stmt.execute();
			
			stmt = conn.prepareStatement(deleteSql);
			stmt.setObject(1, hbaseKey);
			stmt.setObject(2, fromCity);
			stmt.setObject(3, toCity);
			stmt.setObject(4, flightDate);
			stmt.setObject(5, flightNo);
			stmt.execute();
			conn.commit();	
		} catch (SQLException e) {
			try {
				conn.rollback();
			} catch (SQLException e1) {
				throw e1;
			}
			throw e;
		}finally{
	        if (stmt != null) { // 关闭声明  
	            try {  
	                stmt.close();  
	            } catch (SQLException e) {  
	                e.printStackTrace();  
	            }  
	        }  
	    	if(null!=conn){
				try {
					conn.close();
				} catch (SQLException e) {
					e.printStackTrace();
					}
				}
		}
		
	}

	@Override
	public List<SinglePlaneInfoEntity> findByCondition(String hbaseKey,
			String fromCity, String toCity, String flightBatchDate,
			String flightNo) throws SQLException {
		StringBuffer querySQL = new StringBuffer(" select plan.ID , plan.JSON_ID, plan.FLIGHT_DATE, plan.FLIGHTNO,plan.ATTACH_HBASE_KEY,plan.FROM_CITY,plan.TO_CITY  from planeinfo plan where 1=1 ");
		List<Object> conditionVals = new ArrayList<Object>();
		if(null!=hbaseKey && !"".equals(hbaseKey)){
			querySQL.append(" and plan.ATTACH_HBASE_KEY = ? ");
			conditionVals.add(hbaseKey);
		}
		if(null!=fromCity && !"".equals(fromCity)){
			querySQL.append(" and plan.FROM_CITY = ? ");
			conditionVals.add(fromCity);
		}
		if(null!=toCity && !"".equals(toCity)){
			querySQL.append(" and plan.TO_CITY = ? ");
			conditionVals.add(toCity);
		}
		if(null!=flightBatchDate && !"".equals(flightBatchDate)){
			querySQL.append(" and plan.FLIGHT_DATE = ? ");
			conditionVals.add(flightBatchDate);
		}
		if(null!=flightNo && !"".equals(flightNo)){
			querySQL.append(" and plan.FLIGHTNO = ?");
			conditionVals.add(flightNo);
		}
		
		List<SinglePlaneInfoEntity> listEntity =new ArrayList<SinglePlaneInfoEntity>();
		try {
			List<Map<String, Object>> list =DataBaseFetlUtils.getMapList(querySQL.toString(), conditionVals.toArray());
			if(null!=list && list.size()>0){
				for(Map<String, Object> map :list){
					SinglePlaneInfoEntity entity =new SinglePlaneInfoEntity();
					entity.setId(null==map.get("ID")?null:(Long)map.get("ID"));
					entity.setJsonId(null==map.get("JSON_ID")?null:(Long)map.get("JSON_ID"));
					entity.setFlightNo(null==map.get("FLIGHTNO")?null:map.get("FLIGHTNO").toString());
					entity.setFlightDate(null==map.get("FLIGHT_DATE")?null:map.get("FLIGHT_DATE").toString());
					entity.setAttachHbaseKey(null==map.get("ATTACH_HBASE_KEY")?null:map.get("ATTACH_HBASE_KEY").toString());
					entity.setFromCity(null==map.get("FROM_CITY")?null:map.get("FROM_CITY").toString());
					entity.setToCity(null==map.get("TO_CITY")?null:map.get("TO_CITY").toString());
					listEntity.add(entity);
				}
			}
		} catch (SQLException e) {
			throw e;
		}
		
		return listEntity;
	}
	
	@Override
	public List<SinglePlaneInfoEntity> findPlaneInfoProperty(String hbaseKey,String carrierKey,
			String fromCity, String toCity, String flightBatchDate,String flightNo,String grabDate){
		StringBuffer querySQL =new StringBuffer(" select ID as id,SOURCE_ROWKEY as sourceRowkey,ATTACH_HBASE_KEY as attachHbaseKey,CREATE_TIME as createTime,"
			+" GRAB_DATE as grabDate,GRAB_CHANNEL_ID as grabChannelId,GRAB_CHANNEL_NAME as grabChannelName,CARRIER_KEY as carrierKey,CARRIER_NAME as carrierName,"
				+"CARRIER_FULL_NAME as carrierFullName,FLIGHTNO as flightNo,ACTUALLY_FLIGHT_NO as actuallyFlightNo,FLIGHT_TYPE as flightType,"
			+"FLIGHT_DATE as flightDate,START_TIME  as startTime,END_TIME as endTime,FROM_CITY as fromCity,FROM_CITY_NAME as fromCityName,"
				+"TO_CITY as toCity,TO_CITY_NAME as toCityName,AREA_CODE as areaCode,AREA_NAME as areaName,TOTAL_HIGHEST_PRICE as totalHighestPrice,"
			+"TOTAL_LOWEST_PRICE as totalLowestPrice,AGENT_NAME as agentName , JSON_ID as jsonId "+" from planeinfo plan where 1=1 ");
		List<Object> conditionVals = new ArrayList<Object>();
		if(null!=hbaseKey && !"".equals(hbaseKey)){
			querySQL.append(" and plan.ATTACH_HBASE_KEY = ? ");
			conditionVals.add(hbaseKey);
		}
		if(null!=carrierKey && !"".equals(carrierKey)){
			querySQL.append(" and plan.CARRIER_KEY = ? ");
			conditionVals.add(carrierKey);
		}
		if(null!=fromCity && !"".equals(fromCity)){
			querySQL.append(" and plan.FROM_CITY = ? ");
			conditionVals.add(fromCity);
		}
		if(null!=toCity && !"".equals(toCity)){
			querySQL.append(" and plan.TO_CITY = ? ");
			conditionVals.add(toCity);
		}
		if(null!=flightBatchDate && !"".equals(flightBatchDate)){
			querySQL.append(" and plan.FLIGHT_DATE = ? ");
			conditionVals.add(flightBatchDate);
		}
		if(null!=flightNo && !"".equals(flightNo)){
			querySQL.append(" and plan.FLIGHTNO = ?");
			conditionVals.add(flightNo);
		}
		if(null!=grabDate && !"".equals(grabDate)){
			Date start=DateUtils.parse(grabDate, "yyyyMMddHH");
			Calendar calendar=Calendar.getInstance();
			calendar.setTime(start);
			calendar.add(Calendar.HOUR_OF_DAY, 1);
			Date end =calendar.getTime();
			querySQL.append(" and plan.GRAB_DATE between ? and ? ");
			conditionVals.add(start);
			conditionVals.add(end);
		}
		
		List<SinglePlaneInfoEntity> listEntity =new ArrayList<SinglePlaneInfoEntity>();
		listEntity=DataBaseFetlUtils.getBeanList(querySQL.toString(), SinglePlaneInfoEntity.class, conditionVals.toArray());
		return listEntity;
	}

	@Override
	public List<SinglePlaneInfoEntity> findPlanInfoAndFamily(String hbaseKey,
			String fromCity, String toCity, String flightBatchDate,
			String flightNo) {
		StringBuffer querySQL =new StringBuffer(" select plan.ID as id,SOURCE_ROWKEY as sourceRowkey,ATTACH_HBASE_KEY as attachHbaseKey,CREATE_TIME as createTime,"
				+" GRAB_DATE as grabDate,GRAB_CHANNEL_ID as grabChannelId,GRAB_CHANNEL_NAME as grabChannelName,CARRIER_KEY as carrierKey,CARRIER_NAME as carrierName,"
					+"CARRIER_FULL_NAME as carrierFullName,FLIGHTNO as flightNo,ACTUALLY_FLIGHT_NO as actuallyFlightNo,FLIGHT_TYPE as flightType,"
				+"FLIGHT_DATE as flightDate,START_TIME  as startTime,END_TIME as endTime,FROM_CITY as fromCity,FROM_CITY_NAME as fromCityName,"
					+"TO_CITY as toCity,TO_CITY_NAME as toCityName,AREA_CODE as areaCode,AREA_NAME as areaName,TOTAL_HIGHEST_PRICE as totalHighestPrice,"
				+"TOTAL_LOWEST_PRICE as totalLowestPrice,AGENT_NAME as agentName ,JSON_ID as jsonId "
					+"from planeinfo  plan  where 1=1 ");
			List<Object> conditionVals = new ArrayList<Object>();
			if(null!=hbaseKey && !"".equals(hbaseKey)){
				querySQL.append(" and plan.ATTACH_HBASE_KEY = ? ");
				conditionVals.add(hbaseKey);
			}
			if(null!=fromCity && !"".equals(fromCity)){
				querySQL.append(" and plan.FROM_CITY = ? ");
				conditionVals.add(fromCity);
			}
			if(null!=toCity && !"".equals(toCity)){
				querySQL.append(" and plan.TO_CITY = ? ");
				conditionVals.add(toCity);
			}
			if(null!=flightBatchDate && !"".equals(flightBatchDate)){
				querySQL.append(" and plan.FLIGHT_DATE = ? ");
				conditionVals.add(flightBatchDate);
			}
			if(null!=flightNo && !"".equals(flightNo)){
				querySQL.append(" and plan.FLIGHTNO = ?");
				conditionVals.add(flightNo);
			}
			
			List<SinglePlaneInfoEntity> listEntity =new ArrayList<SinglePlaneInfoEntity>();
			listEntity=DataBaseFetlUtils.getBeanList(querySQL.toString(), SinglePlaneInfoEntity.class, conditionVals.toArray());
			if(null!=listEntity && listEntity.size()>0){
				String sql =" select ID AS id , AGENTS_JSON AS agentsJson,CABINS_JSON AS cabinsJson,TRANSITS_JSON AS transitsJson, STOPOVERS_JSON AS stopOversJson from planeinfo_json where ID = ? ";
				for(SinglePlaneInfoEntity entity : listEntity){
					List<PlaneInfoJsonEntity> listJsonEntity = DataBaseFetlUtils.getBeanList(sql, PlaneInfoJsonEntity.class, new Object[]{entity.getJsonId()});
					if(null!=listJsonEntity && listJsonEntity.size()>0){
						entity.setPlaneInfoJsonEntity(listJsonEntity.get(0));
					}
				}
			}
			return listEntity;
	}

	@Override
	public void deleteFormPlaneDay(SinglePlaneInfoEntity entity) throws SQLException {
		Connection conn =null;
		PreparedStatement stmt = null; 
		String deleteSql=" delete from line_day where ID = ? ";
		 try {
			conn = DataBaseFetlUtils.getConnection();
			conn.setAutoCommit(false);
			stmt = conn.prepareStatement(deleteSql);
			stmt.setObject(1, entity.getId());
			stmt.execute();
			conn.commit();	
		} catch (SQLException e) {
			try {
				conn.rollback();
			} catch (SQLException e1) {
				throw e1;
			}
			throw e;
		}finally{
	        if (stmt != null) { // 关闭声明  
	            try {  
	                stmt.close();  
	            } catch (SQLException e) {  
	                e.printStackTrace();  
	            }  
	        }  
	    	if(null!=conn){
				try {
					conn.close();
				} catch (SQLException e) {
					e.printStackTrace();
					}
				}
		}
		
		
	}

	@Override
	public void creatToPlaneDay(SinglePlaneInfoEntity entity) throws SQLException {
		StringBuffer insetSql =new StringBuffer(" insert into line_day(");
		List<Object> conditionVals = new ArrayList<Object>();
		if(null!=entity.getSourceRowkey()){
			insetSql.append("SOURCE_ROWKEY,");
			conditionVals.add(entity.getSourceRowkey());
		}
		if(null!=entity.getAttachHbaseKey()){
			insetSql.append("ATTACH_HBASE_KEY,");
			conditionVals.add(entity.getAttachHbaseKey());
		}
		if(null!=entity.getCreateTime()){
			insetSql.append("CREATE_TIME,");
			conditionVals.add(entity.getCreateTime());
		}
		if(null!=entity.getGrabDate()){
			insetSql.append("GRAB_DATE,");
			conditionVals.add(entity.getGrabDate());
		}
		
		if(null!=entity.getGrabChannelId()){
			insetSql.append("GRAB_CHANNEL_ID,");
			conditionVals.add(entity.getGrabChannelId());
		}
		if(null!=entity.getGrabChannelName()){
			insetSql.append("GRAB_CHANNEL_NAME,");
			conditionVals.add(entity.getGrabChannelName());
		}
		if(null!=entity.getCarrierKey()){
			insetSql.append("CARRIER_KEY,");
			conditionVals.add(entity.getCarrierKey());
		}
		if(null!=entity.getCarrierName()){
			insetSql.append("CARRIER_NAME,");
			conditionVals.add(entity.getCarrierName());
		}
		if(null!=entity.getCarrierFullName()){
			insetSql.append("CARRIER_FULL_NAME,");
			conditionVals.add(entity.getCarrierFullName());
		}
		if(null!=entity.getFlightNo()){
			insetSql.append("FLIGHTNO,");
			conditionVals.add(entity.getFlightNo());
		}
		if(null!=entity.getActuallyFlightNo()){
			insetSql.append("ACTUALLY_FLIGHT_NO,");
			conditionVals.add(entity.getActuallyFlightNo());
		}
		if(null!=entity.getFlightType()){
			insetSql.append("FLIGHT_TYPE,");
			conditionVals.add(entity.getFlightType());
		}
		if(null!=entity.getFlightDate()){
			insetSql.append("FLIGHT_DATE,");
			conditionVals.add(entity.getFlightDate());
		}
		if(null!=entity.getStartTime()){
			insetSql.append("START_TIME,");
			conditionVals.add(entity.getStartTime());
		}
		if(null!=entity.getEndTime()){
			insetSql.append("END_TIME,");
			conditionVals.add(entity.getEndTime());
		}
		if(null!=entity.getFromCity()){
			insetSql.append("FROM_CITY,");
			conditionVals.add(entity.getFromCity());
		}
		if(null!=entity.getFromCityName()){
			insetSql.append("FROM_CITY_NAME,");
			conditionVals.add(entity.getFromCityName());
		}
		if(null!=entity.getToCity()){
			insetSql.append("TO_CITY,");
			conditionVals.add(entity.getToCity());
		}
		if(null!=entity.getToCityName()){
			insetSql.append("TO_CITY_NAME,");
			conditionVals.add(entity.getToCityName());
		}
		if(null!=entity.getAreaCode()){
			insetSql.append("AREA_CODE,");
			conditionVals.add(entity.getAreaCode());
		}
		if(null!=entity.getAreaName()){
			insetSql.append("AREA_NAME,");
			conditionVals.add(entity.getAreaName());
		}
		if(null!=entity.getTotalHighestPrice()){
			insetSql.append("TOTAL_HIGHEST_PRICE,");
			conditionVals.add(entity.getTotalHighestPrice());
		}
		if(null!=entity.getTotalLowestPrice()){
			insetSql.append("TOTAL_LOWEST_PRICE,");
			conditionVals.add(entity.getTotalLowestPrice());
		}
		if(null!=entity.getAgentName()){
			insetSql.append("AGENT_NAME,");
			conditionVals.add(entity.getAgentName());
		}
		
		Connection conn =null;
		PreparedStatement stmt = null;  
		try {
				conn=DataBaseFetlUtils.getConnection();
			 	conn.setAutoCommit(false); 
				StringBuffer sql=new StringBuffer(insetSql.substring(0, insetSql.length()-1));
				String sql2=null;
				sql.append(") values( ");
				if(null!=conditionVals && !conditionVals.isEmpty() && conditionVals.size()>0){
					for(int i=0;i<conditionVals.size();i++){
						sql.append("?,");
					}
					 sql2=sql.substring(0,sql.length()-1)+")";
					 stmt=conn.prepareStatement(sql2);
					 for (int i = 0; i < conditionVals.size(); i++) {  
			                stmt.setObject(i + 1, conditionVals.get(i));  
			            } 
					 stmt.executeUpdate(); 
					
				}
				
				 conn.commit();
				
		} catch (SQLException e) {
				try {
					conn.rollback();
				} catch (SQLException e1) {
					e1.printStackTrace();
				}
		    throw e;
		}finally{
	        if (stmt != null) { // 关闭声明  
	            try {  
	                stmt.close();  
	            } catch (SQLException e) {  
	                e.printStackTrace();  
	            }  
	        }  
	    	if(null!=conn){
				try {
					conn.close();
				} catch (SQLException e) {
					e.printStackTrace();
					}
				}
		}
		
	}

	@Override
	public List<SinglePlaneInfoEntity> findFromPlaneDay(String hbaseKey,
			String fromCity, String toCity, String flightBatchDate) throws SQLException {
		StringBuffer querySQL = new StringBuffer(" select plan.ID, plan.FLIGHT_DATE, plan.FROM_CITY,plan.TO_CITY, plan.TOTAL_LOWEST_PRICE  from line_day plan where 1=1 ");
		List<Object> conditionVals = new ArrayList<Object>();
		if(null!=hbaseKey && !"".equals(hbaseKey)){
			querySQL.append(" and plan.ATTACH_HBASE_KEY = ? ");
			conditionVals.add(hbaseKey);
		}
		if(null!=fromCity && !"".equals(fromCity)){
			querySQL.append(" and plan.FROM_CITY = ? ");
			conditionVals.add(fromCity);
		}
		if(null!=toCity && !"".equals(toCity)){
			querySQL.append(" and plan.TO_CITY = ? ");
			conditionVals.add(toCity);
		}
		if(null!=flightBatchDate && !"".equals(flightBatchDate)){
			querySQL.append(" and plan.FLIGHT_DATE = ? ");
			conditionVals.add(flightBatchDate);
		}
	
		List<SinglePlaneInfoEntity> listEntity =new ArrayList<SinglePlaneInfoEntity>();
		try {
			List<Map<String, Object>> list =DataBaseFetlUtils.getMapList(querySQL.toString(), conditionVals.toArray());
			if(null!=list && list.size()>0){
				for(Map<String, Object> map :list){
					SinglePlaneInfoEntity entity =new SinglePlaneInfoEntity();
					entity.setId(null==map.get("ID")?null:(Long)map.get("ID"));
					entity.setFromCity(map.get("FROM_CITY").toString());
					entity.setToCity(map.get("TO_CITY").toString());
					entity.setFlightDate(map.get("FLIGHT_DATE").toString());
					entity.setTotalLowestPrice(null==map.get("TOTAL_LOWEST_PRICE")?null:(Double)map.get("TOTAL_LOWEST_PRICE"));
					listEntity.add(entity);
				}
			}
		} catch (SQLException e) {
			throw e;
		}
		
		return listEntity;
	}

	@Override
	public List<SinglePlaneInfoEntity> findByGrabDate(String hbaseKey,
			String fromCity, String toCity, String grabDate) {
		StringBuffer querySQL =new StringBuffer(" select plan.ID as id,SOURCE_ROWKEY as sourceRowkey,ATTACH_HBASE_KEY as attachHbaseKey,CREATE_TIME as createTime,"
				+" GRAB_DATE as grabDate,GRAB_CHANNEL_ID as grabChannelId,GRAB_CHANNEL_NAME as grabChannelName,CARRIER_KEY as carrierKey,CARRIER_NAME as carrierName,"
					+"CARRIER_FULL_NAME as carrierFullName,FLIGHTNO as flightNo,ACTUALLY_FLIGHT_NO as actuallyFlightNo,FLIGHT_TYPE as flightType,"
				+"FLIGHT_DATE as flightDate,START_TIME  as startTime,END_TIME as endTime,FROM_CITY as fromCity,FROM_CITY_NAME as fromCityName,"
					+"TO_CITY as toCity,TO_CITY_NAME as toCityName,AREA_CODE as areaCode,AREA_NAME as areaName,TOTAL_HIGHEST_PRICE as totalHighestPrice,"
				+"TOTAL_LOWEST_PRICE as totalLowestPrice,AGENT_NAME as agentName ,JSON_ID as jsonId "
					+"from planeinfo  plan  where 1=1 ");
			List<Object> conditionVals = new ArrayList<Object>();
			if(null!=hbaseKey && !"".equals(hbaseKey)){
				querySQL.append(" and plan.ATTACH_HBASE_KEY = ? ");
				conditionVals.add(hbaseKey);
			}
			if(null!=fromCity && !"".equals(fromCity)){
				querySQL.append(" and plan.FROM_CITY = ? ");
				conditionVals.add(fromCity);
			}
			if(null!=toCity && !"".equals(toCity)){
				querySQL.append(" and plan.TO_CITY = ? ");
				conditionVals.add(toCity);
			}
			if(null!=grabDate && !"".equals(grabDate)){
				Date start=DateUtils.parse(grabDate, "yyyyMMddHH");
				Calendar calendar=Calendar.getInstance();
				calendar.setTime(start);
				calendar.add(Calendar.HOUR_OF_DAY, 1);
				Date end =calendar.getTime();
				querySQL.append(" and plan.GRAB_DATE between ? and ? ");
				conditionVals.add(start);
				conditionVals.add(end);
			}
			
			
			List<SinglePlaneInfoEntity> listEntity =new ArrayList<SinglePlaneInfoEntity>();
			listEntity=DataBaseFetlUtils.getBeanList(querySQL.toString(), SinglePlaneInfoEntity.class, conditionVals.toArray());
			if(null!=listEntity && listEntity.size()>0){
				String sql =" select ID AS id , AGENTS_JSON AS agentsJson,CABINS_JSON AS cabinsJson,TRANSITS_JSON AS transitsJson, STOPOVERS_JSON AS stopOversJson from planeinfo_json where ID = ? ";
				for(SinglePlaneInfoEntity entity : listEntity){
					List<PlaneInfoJsonEntity> listJsonEntity = DataBaseFetlUtils.getBeanList(sql, PlaneInfoJsonEntity.class, new Object[]{entity.getJsonId()});
					if(null!=listJsonEntity && listJsonEntity.size()>0){
						entity.setPlaneInfoJsonEntity(listJsonEntity.get(0));
					}
				}
			}
			return listEntity;
	}

	@Override
	public void create(Collection<SinglePlaneInfoEntity> set)
			throws SQLException {
		Connection conn =null;
		PreparedStatement stmt = null;  
		PreparedStatement tsmtJson=null;
		StringBuffer insetSql =new StringBuffer(1024*1024*2);
		insetSql.append(" insert into planeInfo(SOURCE_ROWKEY,ATTACH_HBASE_KEY,CREATE_TIME,GRAB_DATE,GRAB_CHANNEL_ID," +
				"GRAB_CHANNEL_NAME,CARRIER_KEY,CARRIER_NAME,CARRIER_FULL_NAME,FLIGHTNO,ACTUALLY_FLIGHT_NO,FLIGHT_TYPE,FLIGHT_DATE," +
				"START_TIME,END_TIME,FROM_CITY,FROM_CITY_NAME,TO_CITY,TO_CITY_NAME,AREA_CODE,AREA_NAME,TOTAL_HIGHEST_PRICE," +
				"TOTAL_LOWEST_PRICE,AGENT_NAME,JSON_ID,TOTAL_HIGHEST_TAXES_PRICE,TOTAL_LOWEST_TAXES_PRICE,SUM_HIGHEST_PRICE,SUM_LOWEST_PRICE) values");
		
		StringBuffer insetJsonSql =new StringBuffer(1024*1024*2);
		insetJsonSql.append("insert into planeinfo_json(ID,AGENTS_JSON,CABINS_JSON,TRANSITS_JSON,STOPOVERS_JSON) values");
		Date grabTime=null;
		List<String> keys= new ArrayList<String>();
		for(SinglePlaneInfoEntity entity:  set){
	    	String jsonString=null;
	    	PlaneInfoJsonEntity planeInfoJsonEntity=new PlaneInfoJsonEntity();
			 if(null!=entity.getAgents() && entity.getAgents().size()>0 ){
					jsonString =JsonUtil.getJSONString(entity.getAgents());
					planeInfoJsonEntity.setAgentsJson(jsonString);
			 }
			 if(null!=entity.getCabins()&&entity.getCabins().size()>0){
					jsonString =JsonUtil.getJSONString(entity.getCabins());
					planeInfoJsonEntity.setCabinsJson(jsonString);
			 } 
			 if(null!=entity.getTransits()&& entity.getTransits().size()>0){
				 jsonString =JsonUtil.getJSONString(entity.getTransits());
				 planeInfoJsonEntity.setTransitsJson(jsonString);
			 }
			 if(null!=entity.getStopOvers()&& entity.getStopOvers().size()>0){
				 jsonString =JsonUtil.getJSONString(entity.getStopOvers());
				 planeInfoJsonEntity.setStopOversJson(jsonString);
			 }
			 if(null!=jsonString){
				 long id=RandomUtils.getAtomicCounter();
				 entity.setJsonId(id);
				 planeInfoJsonEntity.setId(id);
				 entity.setPlaneInfoJsonEntity(planeInfoJsonEntity); 
			 }
	    }
		
	    PlaneInfoJsonEntity planeInfoJsonEntity=null;
	    for(SinglePlaneInfoEntity entity:  set){
	    	planeInfoJsonEntity=entity.getPlaneInfoJsonEntity();
	    	if(null==grabTime)
	    		grabTime=entity.getGrabDate();
	    	keys.add(entity.getRowkey());
	    	
	    	if(null!=planeInfoJsonEntity){
	    		insetJsonSql.append("(").append(planeInfoJsonEntity.getId()).append(",").
	    		append(null==planeInfoJsonEntity.getAgentsJson()?null:"'"+planeInfoJsonEntity.getAgentsJson().replace("\"", "\\"+"\"")+"'").append(",").
	    		append(null==planeInfoJsonEntity.getCabinsJson()?null:"'"+planeInfoJsonEntity.getCabinsJson().replace("\"", "\\"+"\"")+"'").append(",").
	    		append(null==planeInfoJsonEntity.getTransitsJson()?null:"'"+planeInfoJsonEntity.getTransitsJson().replace("\"", "\\"+"\"")+"'").append(",").
	    		append(null==planeInfoJsonEntity.getStopOversJson()?null:"'"+planeInfoJsonEntity.getStopOversJson().replace("\"", "\\"+"\"")+"'").append("),");
	    	}
	    	
	    	insetSql.append("(").append(null==entity.getSourceRowkey()?null:"'"+entity.getSourceRowkey()+"'").append(",")
	    			.append(null==entity.getAttachHbaseKey()?null:"'"+entity.getAttachHbaseKey()+"'").append(",")
	    			.append(null==entity.getCreateTime()?null:"'"+DateUtils.format(entity.getCreateTime(), "yyyy-MM-dd HH:mm:ss")+"'").append(",")
	    			.append(null==entity.getGrabDate()?null:"'"+DateUtils.format(entity.getGrabDate(), "yyyy-MM-dd HH:mm:ss")+"'").append(",")
	    			.append(entity.getGrabChannelId()).append(",")
	    			.append(null==entity.getGrabChannelName()?null:"'"+entity.getGrabChannelName()+"'").append(",")
	    			.append(null==entity.getCarrierKey()?null:"'"+entity.getCarrierKey()+"'").append(",")
	    			.append(null==entity.getCarrierName()?null:"'"+entity.getCarrierName()+"'").append(",")
	    			.append(null==entity.getCarrierFullName()?null:"'"+entity.getCarrierFullName()+"'").append(",")
	    			.append(null==entity.getFlightNo()?null:"'"+entity.getFlightNo()+"'").append(",")
	    			.append(null==entity.getActuallyFlightNo()?null:"'"+entity.getActuallyFlightNo()+"'").append(",")
	    			.append(null==entity.getFlightType()?null:"'"+entity.getFlightType()+"'").append(",")
	    			.append(null==entity.getFlightDate()?null:"'"+entity.getFlightDate()+"'").append(",")
	    			.append(null==entity.getStartTime()?null:"'"+DateUtils.format(entity.getStartTime(), "yyyy-MM-dd HH:mm:ss")+"'").append(",")
	    			.append(null==entity.getEndTime()?null:"'"+DateUtils.format(entity.getEndTime(), "yyyy-MM-dd HH:mm:ss")+"'").append(",")
	    			.append(null==entity.getFromCity()?null:"'"+entity.getFromCity()+"'").append(",")
	    			.append(null==entity.getFromCityName()?null:"'"+entity.getFromCityName()+"'").append(",")
	    			.append(null==entity.getToCity()?null:"'"+entity.getToCity()+"'").append(",")
	    			.append(null==entity.getToCityName()?null:"'"+entity.getToCityName()+"'").append(",")
	    			.append(null==entity.getAreaCode()?null:"'"+entity.getAreaCode()+"'").append(",")
	    			.append(null==entity.getAreaName()?null:"'"+entity.getAreaName()+"'").append(",")
	    			.append(entity.getTotalHighestPrice()).append(",")
	    			.append(entity.getTotalLowestPrice()).append(",")
	    			.append(null==entity.getAgentName()?null:"'"+entity.getAgentName()+"'").append(",")
	    			.append(entity.getJsonId()).append(",")
	    			.append(entity.getTotalHighestTaxesPrice()).append(",")
	    			.append(entity.getTotalLowestTaxesPrice()).append(",")
	    			.append(entity.getSumHighestPrice()).append(",")
	    			.append(entity.getSumLowestPrice())
	    			.append("),");
		 }
	    insetJsonSql.delete(insetJsonSql.length()-1, insetJsonSql.length());
	    insetSql.delete(insetSql.length()-1, insetSql.length());
	    
	    try{
	    	DataBaseFetlUtils dbUtils=new DataBaseFetlUtils();
	    	conn=dbUtils.getConn();
	    	try {
	  			conn.setAutoCommit(false);
	  		} catch (Exception e) {
	  			conn=dbUtils.getConn();
	  		} 
	    	tsmtJson=conn.prepareStatement(insetJsonSql.toString());
	    	stmt=conn.prepareStatement(insetSql.toString());
	    	if(null!=planeInfoJsonEntity){
	    		tsmtJson.execute();
	    	}
	    	stmt.execute();
	    	conn.commit();		
	    	keys.clear();
	    }catch(SQLException e){
	    	conn.rollback();
	    	log.error(StringUtils.getStackTrace(e));
	    	FetlErrorMessage vo=FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.last, this.getClass().getName(),DateUtils.format(grabTime, "yyyyMMddHH"), e);
			vo.setKeys(keys);
			FetlErrorUtils.saveToHbase(vo);
			keys.clear();
	    	throw e;
	    }finally{
	    	if(null!=insetJsonSql)
	    		insetJsonSql.delete(0, insetJsonSql.length());
	    	if(null!=insetSql)
	    		insetSql.delete(0, insetSql.length());
	    	  if (stmt != null) { // 关闭声明  
		            try {  
		                stmt.close();  
		            } catch (SQLException e) {  
		            	log.error(StringUtils.getStackTrace(e));
		            }  
		        }  
	    	  if(tsmtJson!=null){
	    		  tsmtJson.close();
	    	  }
	    	if(null!=conn){
				try {
					conn.close();
				} catch (SQLException e) {
					log.error(StringUtils.getStackTrace(e));
					}
				}
	    }
		
	}

	@Override
	public void delete(Collection<SinglePlaneInfoEntity> oldList)
			throws SQLException {
		Connection conn =null;
		PreparedStatement stmt = null; 
		PreparedStatement stmtJson = null;
		String deletejsonSql= " delete from planeinfo_json where ID=? ";
		String deleteSql=" delete from planeInfo where ID = ? ";
		//conn = DataBaseFetlUtils.getConnection();
		DataBaseFetlUtils dbUtils=new DataBaseFetlUtils();
    	conn=dbUtils.getConn();
		try {
			long l=0;
			stmtJson = conn.prepareStatement(deletejsonSql);
			stmt = conn.prepareStatement(deleteSql);
			 if(!conn.isValid(0)){
	    			conn=DataBaseFetlUtils.getConnection();
	    		}
			conn.setAutoCommit(false);
			for(SinglePlaneInfoEntity entity:oldList){
				 try {
					if (null != entity.getJsonId() && entity.getJsonId()>0) {
						stmtJson.setObject(1, entity.getJsonId());
						stmtJson.addBatch();
					//	stmt.execute();
					}
					stmt.setObject(1, entity.getId());
					stmt.addBatch();
					//stmt.execute();
					l++;
					if (l % 1000 == 0 ) {
						stmtJson.executeBatch();
						stmt.executeBatch();
						 conn.commit();	
					 }
				} catch (SQLException e) {
					log.error(StringUtils.getStackTrace(e));
					try {
						conn.rollback();
					} catch (SQLException e1) {
						log.error(StringUtils.getStackTrace(e));
					}
				}
			
			}
			stmtJson.executeBatch();
			stmt.executeBatch();
			conn.commit();	
		} catch (Exception e) {
			e.printStackTrace();
			try {
				throw e;
			} catch (Exception e1) {
				// TODO Auto-generated catch block
				e1.printStackTrace();
			}
		}finally{
			if (stmt != null) { // 关闭声明  
		        try {  
		            stmt.close();  
		        } catch (SQLException e) {  
		            e.printStackTrace();  
		        }  
		    } 
			if(null!=conn){
				conn.close();
			}
		}
	}
}
